package com.windy.kafka.trace.config;


import com.windy.common.transmit.kafka.KafkaConsumerTransitHeader;
import com.windy.common.transmit.kafka.KafkaProducerTransmitHeader;
import com.windy.kafka.trace.annotation.EnableMQ;
import com.windy.kafka.trace.consumer.DefaultMessageConsumerProcessor;
import com.windy.kafka.trace.consumer.MessageConsumerProcessor;
import com.windy.kafka.trace.interceptor.AnnotationMessageExecutionInterceptor;
import com.windy.kafka.trace.interceptor.KafkaListenerAspect;
import com.windy.kafka.trace.producer.DefaultMessageSendProcessor;
import com.windy.kafka.trace.producer.MessageSendProcessor;
import com.windy.kafka.trace.producer.TracingProducerFactory;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.DefaultKafkaProducerFactoryCustomizer;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Role;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;

import java.lang.annotation.Annotation;
import java.util.List;

/**
 * Created by liyufeng on 2022/3/22
 */
@Configuration(proxyBeanMethods = false)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class ProxyMessageSenderConfiguration extends AbstractSenderConfiguration implements BeanFactoryAware {

    private BeanFactory beanFactory;

    @Bean
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public MessageAnnotationBeanPostProcessor messageAdvisor() {
        Assert.notNull(this.enableMQ, "@EnableMQ annotation metadata was not injected");
        MessageAnnotationBeanPostProcessor mbp = new MessageAnnotationBeanPostProcessor();
        Class<? extends Annotation> customAsyncAnnotation = this.enableMQ.getClass("annotation");
        if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableMQ.class, "annotation")) {
            mbp.setAsyncAnnotationType(customAsyncAnnotation);
        }
        mbp.setAdvice(messageExecutionInterceptor());
        mbp.setProxyTargetClass(this.enableMQ.getBoolean("proxyTargetClass"));
        mbp.setOrder(this.enableMQ.<Integer>getNumber("order"));
        return mbp;
    }

    @Bean
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public MessageSendProcessor sendPostProcessor() {
        return new DefaultMessageSendProcessor();
    }

    @Bean
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public MessageConsumerProcessor messageConsumerProcessor(){
        return new DefaultMessageConsumerProcessor();
    }

    @Bean
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public AnnotationMessageExecutionInterceptor messageExecutionInterceptor() {
        KafkaTemplate kafkaTemplate = beanFactory.getBean("kafkaTemplate", KafkaTemplate.class);
        AnnotationMessageExecutionInterceptor interceptor = new AnnotationMessageExecutionInterceptor(kafkaTemplate);
        if (this.executor != null) {
            interceptor.setExecutor(this.executor);
        }
        if (this.exceptionHandler != null) {
            interceptor.setExceptionHandler(this.exceptionHandler);
        }
        return interceptor;
    }

    @Bean
    public KafkaProducerTransmitHeader kafkaProducerTransmitHeader(){
        return new KafkaProducerTransmitHeader();
    }

    @Bean
    public KafkaConsumerTransitHeader kafkaConsumerTransitHeader(){
        return new KafkaConsumerTransitHeader();
    }

    @Bean
    public KafkaListenerAspect kafkaListenerAspect( List<MessageConsumerProcessor>messageConsumerProcessors){
        return new KafkaListenerAspect(kafkaConsumerTransitHeader(),messageConsumerProcessors);
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }


    @Configuration(proxyBeanMethods = false)
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    @AutoConfigureBefore(KafkaAutoConfiguration.class)
    @EnableConfigurationProperties(KafkaProperties.class)
    public static class kafkaConfiguration {
        private final KafkaProperties properties;

        public kafkaConfiguration(KafkaProperties properties) {
            this.properties = properties;
        }

        @Bean
        @ConditionalOnMissingBean(ProducerFactory.class)
        public TracingProducerFactory producerFactory(ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers,
                                                      KafkaProducerTransmitHeader kafkaProducerTransmitHeader,
                                                      ObjectProvider<List<MessageSendProcessor>>messageSendProcessors  ) {
            DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(
                    this.properties.buildProducerProperties());
            String transactionIdPrefix = this.properties.getProducer().getTransactionIdPrefix();
            if (transactionIdPrefix != null) {
                factory.setTransactionIdPrefix(transactionIdPrefix);
            }
            customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
            return new TracingProducerFactory(factory, messageSendProcessors.getIfAvailable(),kafkaProducerTransmitHeader);
        }

        @Bean
        public ConsumerAwareListenerErrorHandler myConsumerAwareErrorHandler() {
            return new ConsumerAwareListenerErrorHandler() {
                @Override
                public Object handleError(Message<?> message, ListenerExecutionFailedException exception,
                                          Consumer<?, ?> consumer) {
                    System.out.println("--- 发生消费异常 ---");
                    System.out.println(message.getPayload());
                    System.out.println(exception);
                    return null;
                }
            };
        }



//        @Bean
//        @ConditionalOnMissingBean(ConsumerFactory.class)
//        public ConsumerFactory<Object, Object> kafkaConsumerFactory(
//                ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers) {
//            DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(
//                    this.properties.buildConsumerProperties());
//            customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
//            return new TracingConsumerFactory<>(factory,kafkaConsumerTransitHeader());
//        }
//
//        @Bean
//        @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
//        ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
//                ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
//                ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers) {
//            ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
//            configurer.configure(factory, kafkaConsumerFactory(customizers));
//            return factory;
//        }
    }


}
